package com.dingwen.treasure.business.config;

import cn.hutool.core.convert.Convert;
import cn.hutool.extra.spring.SpringUtil;
import com.dingwen.treasure.base.constant.RabbitmqConstant;
import com.dingwen.treasure.base.enums.MsgState;
import com.dingwen.treasure.mybatisplus.service.TreasureMessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;

/**
 * 　RabbitTemplateConfig
 * <p>消息可靠投递</p>
 * <pre>
 *     1. 消息投递到指定交换机时触发回调  {@code  rabbitTemplate.setConfirmCallback}
 *      1.1 yml配置{@yml spring.rabbitmq.publisher-confirm-type: correlated}
 *      1.2 correlated: 消息投递成功，调用
 *      1.3 simple: 兼备correlated的功能，另执行wait等待broker的反馈再进行操作
 *      1.4 none: 关闭此项功能
 *     2. 消息投递到队列之后，之前到操作发生错误进行回调
 *      2.1 yml 配置publisher-returns: true
 *      2.2 开启broker给生产着反馈消息{@code rabbitTemplate.setMandatory(Boolean.TRUE);}
 *      2.3 rabbitTemplate.setReturnsCallback
 *
 *     3. 当mandatory设置为true，会通过Basic.Return RPC命令将消息返回给发布者；当mandatory设置为false时，该消息被直接丢弃
 *
 * </pre>
 * 　@author dingwen
 * 　@date 2022/6/5
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
public class RabbitTemplateConfig {


    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setConfirmCallback(getConfirmCallback());
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(getReturnCallback());
        return rabbitTemplate;
    }


    /**
     * 确保消息发送确认回调 <br>
     * <p>当消息发送到broker分发到指定交换机之后进行此方法到回调</p>
     */
    private RabbitTemplate.ConfirmCallback getConfirmCallback() {
        return (correlationData, isSuccess, errMessage) -> {
            // 消息id
            Assert.notNull(correlationData, "correlationData 不能为空");
            String messageId = correlationData.getId();
            if (!isSuccess || !StringUtils.isNotBlank(messageId)) {
                // 消息投递失败
                log.error("messageId:{}投递失败,由定时任务进行补偿", messageId);
                TreasureMessageService messageService = SpringUtil.getBean(TreasureMessageService.class);
                messageService.updateMessageState(MsgState.SAVED, Convert.toLong(messageId));
            }
        };
    }

    /**
     * 开启ReturnsCallback回调 <br>
     * <p>
     * 消息投递到队列之前出现错误执行回调方法
     * </p>
     */
    private RabbitTemplate.ReturnCallback getReturnCallback() {
        return (Message message, int code, String msg, String exchange, String routingKey) -> {
            TreasureMessageService messageService = SpringUtil.getBean(TreasureMessageService.class);
            messageService.updateMessageState(MsgState.SAVED, Convert.toLong(message.getMessageProperties()
                    .getHeader(RabbitmqConstant.SPRING_RETURNED_MESSAGE_CORRELATION)));
            log.error("消息投递到队列到过程中发生错误，由定时任务进行补偿，消息主体：{},描述码：{},描述：{},交换机：{},路由键：{}", message, code, msg, exchange, routingKey);
        };
    }
}
